package algorithm;

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.*;

/**
 * 平衡性: 是指哈希的结果能够尽可能分布到所有的缓冲中去，这样可以使得所有的缓冲空间都得到利用。
 * 单调性: 是指如果已经有一些数据通过哈希分配到了相应的机器上，又有新的机器加入到系统中。哈希的结果应能够保证原有已分配的内容可以被映射到原有的或者新的机器中去，而不会被映射到旧的机器集合中的其他机器上。
 * 就是原有的数据要么还是呆在它所在的机器上不动，要么被迁移到新的机器上，而不会迁移到旧的其他机器上
 * 分散性:
 * 负载:
 */
public class ConsistentHash<T> {
    private final HashFunction hashFunction;
    //hash值到真实节点的映射
    private final SortedMap<Long, T> circle = new TreeMap<>();

    public ConsistentHash(HashFunction hashFunction, Collection<T> nodes) {
        this.hashFunction = hashFunction;
        for (T node : nodes)
            add(node);
    }

    /*
     * 对于一个实际机器节点 node, 不同的点(i不同)有不同的hash值,但都对应同一个实际机器node
     */
    public void add(T node) {
        circle.put(hashFunction.hash(node.toString()), node);
    }

    public void remove(T node) {
        circle.remove(hashFunction.hash(node.toString()));
    }

    /*
     * 获得一个最近的顺时针节点,根据给定的key 取Hash
     * 然后再取得顺时针方向上最近的一个虚拟节点对应的实际节点
     * 再从实际节点中取得 数据
     */
    public T get(Object key) {
        if (circle.isEmpty())
            return null;
        // node 用String来表示,获得node在哈希环中的hashCode
        long hash = hashFunction.hash((String) key);
        //按顺时针方向寻找机器
        if (!circle.containsKey(hash)) {
            SortedMap<Long, T> tailMap = circle.tailMap(hash);
            hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
        }
        return circle.get(hash);
    }

    public long getSize() {
        return circle.size();
    }

    /*
     * 查看MD5算法生成的hashCode值
     */
    public void testBalance() {
        //获得TreeMap中所有的Key
        Set<Long> sets = circle.keySet();
        //将获得的Key集合排序
        SortedSet<Long> sortedSets = new TreeSet<>(sets);
        for (Long hashCode : sortedSets) {
            System.out.println(hashCode);
        }

        System.out.println("----每个位置的距离: ----");
        /*
         * 查看用MD5算法生成的long hashCode 相邻两个hashCode的差值
         */
        Iterator<Long> it = sortedSets.iterator();
        Iterator<Long> it2 = sortedSets.iterator();
        if (it2.hasNext())
            it2.next();
        long keyPre, keyAfter;
        while (it.hasNext() && it2.hasNext()) {
            keyPre = it.next();
            keyAfter = it2.next();
            System.out.println(keyAfter - keyPre);
        }
    }


    public static void main(String[] args) {
        Set<String> nodes = new HashSet<>();
        nodes.add("NODE_A");
        nodes.add("NODE_B");
        nodes.add("NODE_C");

        ConsistentHash<String> consistentHash = new ConsistentHash<>(new HashFunction(), nodes);
        System.out.println("hash 圆的大小: " + consistentHash.getSize());
        System.out.println("每个节点的  位置: ");
        consistentHash.testBalance();

        System.out.println("---------------测试数据位置-----------------");
        int isA = 0;
        int isB = 0;
        int isC = 0;
        for (int i = 0; i < 100000; i++) {
            if ("NODE_A".equals(consistentHash.get(UUID.randomUUID().toString()))) {
                isA++;
            } else if ("NODE_B".equals(consistentHash.get(UUID.randomUUID().toString()))) {
                isB++;
            } else if ("NODE_C".equals(consistentHash.get(UUID.randomUUID().toString()))) {
                isC++;
            }
        }

        System.out.println("A:" + isA);
        System.out.println("B:" + isB);
        System.out.println("C:" + isC);
    }


    /*
     * 实现一致性哈希算法中使用的哈希函数,使用MD5算法来保证一致性哈希的平衡性 hashcode
     */
    static class HashFunction {
        private MessageDigest md5 = null;

        public long hash(String key) {
            if (md5 == null) {
                try {
                    md5 = MessageDigest.getInstance("MD5");
                } catch (NoSuchAlgorithmException e) {
                    e.printStackTrace();
                }
            }

            md5.reset();
            md5.update(key.getBytes());
            byte[] bKey = md5.digest();
            //具体的哈希函数实现细节--每个字节 & 0xFF 再移位
            long result = ((long) (bKey[3] & 0xFF) << 24)
                    | ((long) (bKey[2] & 0xFF) << 16
                    | ((long) (bKey[1] & 0xFF) << 8) | (long) (bKey[0] & 0xFF));
            return result & 0xffffffffL;
        }
    }
}